package com.atguigu.realtime.apps

import com.alibaba.fastjson.JSON
import com.atguigu.realtime.beans.StartLog
import com.atguigu.realtime.constants.{PrefixConstant, TopicConstant}
import com.atguigu.realtime.utils.{DStreamUtil, DateTimeUtil, JedisUtil}
import org.apache.hadoop.hbase.HBaseConfiguration
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{CanCommitOffsets, HasOffsetRanges, OffsetRange}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import redis.clients.jedis.Jedis

/**
 * Created by Smexy on 2022/8/26
 *
 *    at least once + 幂等输出(hbase) 实现 精确一次
 */
object DAUApp extends BaseApp {

  override var batchDuration: Int = 10
  override var appName: String = "DAUApp"
  override var groupId: String = "220409realtime"
  //消费startlog
  override var topic: String = TopicConstant.STARTUP_LOG

  def parseRecordToBean(rdd: RDD[ConsumerRecord[String, String]]):RDD[StartLog] = {

    //每一条 ConsumerRecord，解析为一个 StartLog
    rdd.map(record => {

      //只有从kafka中拿到的字段，自定义的字段没有赋值
      val startLog: StartLog = JSON.parseObject(record.value(), classOf[StartLog])

      //为自定义的字段赋值
      startLog.start_date = DateTimeUtil.parseMillTsToDate(startLog.ts.toLong)
      startLog.start_time = DateTimeUtil.parseMillTsToDateTime(startLog.ts.toLong)

      startLog.id = startLog.start_time + "_" + startLog.mid

      startLog

    })
  }

  /*
      将这个批次中，相同日期和相同mid的  startlog分在一个组中，挑选出组中 ts最小的那条，返回。

          分组 ------->聚合 -------> 每组取 ts最小
   */
  def removeDuplicateLogInCommonBatch(rdd: RDD[StartLog]):RDD[StartLog] = {

    val rdd1: RDD[((String, String), StartLog)] = rdd.map(log => ((log.start_date, log.mid), log))
    val rdd2: RDD[((String, String), StartLog)] = rdd1.reduceByKey((log1: StartLog, log2: StartLog) => {
      if (log1.ts <= log2.ts) {
        log1
      } else {
        log2
      }
    })
    val rdd3: RDD[StartLog] = rdd2.values
    rdd3

  }

  /*
      当前的场景是： 后续的批次中的每个 设备，要到redis中来看是否已经有相同 mid 存在，如果已经存在，就过滤。

          如果单值类型的string :    get(mid)            效率高。  时间复杂度 O(1)
                                      不太方便管理。粒度是一天一个设备是一个K-V。
                                      可以设置过期时间，设置key 24h自动过期。

          如果是set类型：           sismember(mid)      效率低。 需要遍历集合，才能判断元素是否在集合中，时间复杂度O(n)

          这里选择string。

              Key:    前缀_日期_mid
              value:  string。 什么都不存


   */
  def removeDuplicateLogFromHistoryBatch(rdd: RDD[StartLog]):RDD[StartLog] = {

    //需要查询redis，因此以分区为单位操作
    rdd.mapPartitions(partition => {

      val jedis: Jedis = JedisUtil.getJedis

      //使用连接  filter 保留 返回true
      val logs: Iterator[StartLog] = partition.filter(log => !jedis.exists(PrefixConstant.dau_redis_Preffix + "_" + log.start_date + "_" + log.mid))

      jedis.close()

      logs

    })

  }

  def main(args: Array[String]): Unit = {

    //重写父类中的context
    context = new StreamingContext("local[*]",appName,Seconds(batchDuration))

    runSparkStreamingApp{

      val ds: InputDStream[ConsumerRecord[String, String]] = DStreamUtil.getDStream(context, groupId, topic)

      ds.foreachRDD(rdd => {

        if (!rdd.isEmpty()){

          //获取偏移量
          val ranges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

          //将ConsumerRecord中的value转换为想计算的bean
          //建模的原则：  kafka中存储的每一个字段都要创建，还可以再加上需要的字段
          val rdd1: RDD[StartLog] = parseRecordToBean(rdd)

          //同批次去重
          val rdd2: RDD[StartLog] = removeDuplicateLogInCommonBatch(rdd1)

          //和redis中已经记录过启动日志的设备比对，只留下当天没有记录过启动日志的
          val rdd3: RDD[StartLog] = removeDuplicateLogFromHistoryBatch(rdd2)

          // 如果引入了 phoenix-spark，提供了现成的方法，供你使用
          // 导入org.apache.phoenix.spark 中的所有方法，可以直接在这个类中使用
          import org.apache.phoenix.spark._

          rdd3.cache()

          println("要写出:"+rdd3.count())
          /*
            tableName: String    phoenix的表名
            cols: Seq[String]   指RDD的T类型的属性按照顺序要写入表中的哪些列?

            conf: Configuration = 有标准写法。HBaseConfiguration.create()
            zkUrl: Option[String] = None  Phoenix 连接的 hbase 所使用的zk地址
           */
          //把启动时间最早的log写入hbase
          rdd3.saveToPhoenix(
            "REALTIME2022_STARTLOG",
            Seq("ID","OPEN_AD_MS","OS","CH","IS_NEW","MID","OPEN_AD_ID","VC","AR",
              "UID","ENTRY","OPEN_AD_SKIP_MS","MD","LOADING_TIME","BA","TS","START_DATE","START_TIME") ,
            HBaseConfiguration.create(),
            Some("hadoop102:2181")
          )

          //把已经记录过启动日志的设备缓存在redis，以供后续的批次进行比对
          rdd3.foreachPartition(partition => {

            val jedis: Jedis = JedisUtil.getJedis

            partition.foreach(log => jedis.setex(PrefixConstant.dau_redis_Preffix + "_" + log.start_date + "_" + log.mid, 60 * 60 * 24,""))

            jedis.close()

          })

          //提交偏移量
          ds.asInstanceOf[CanCommitOffsets].commitAsync(ranges)


        }

      })

    }

  }
}
